--客诉信息
with work_order as  (
    select
        waybill_no,                                         --单号
        collect_set(second_level_name) as second_type_name, --运单所有二级类型名称
        max(is_flas_sign) as is_flas_sign,                  --优先取签收未收到标签
        max(create_time) as work_create_time                --最新投诉时间
    from
        (
            select
                waybill_no,
                second_level_code ,
                second_level_name,
                1  as is_flas_sign,
                create_time
            from jms_dwd.dwd_project_work_order_base_zipper_dt  --平台工单(卖家投诉问题)
            where dt  between date_add('{{ execution_date | cst_ds }}',-60) and  '{{ execution_date | cst_ds }}'
              and work_order_source_id = 1
              and date(create_time) >= date_add('{{ execution_date | cst_ds }}',-60)
union all
select
    waybill_no,
    second_type_code as second_level_code,
    second_type_name as second_level_name,
    if(second_type_code = 2104,1,0) as is_flas_sign,  --是否签收未收到
    create_time
from jms_dwd.dwd_work_order_base_zipper_dt        --普通工单
where dt between date_add('{{ execution_date | cst_ds }}',-60) and  '{{ execution_date | cst_ds }}'
  and date(create_time) >= date_add('{{ execution_date | cst_ds }}',-60)
    )t group by waybill_no
    ),
--退转信息
    reback as (
select
    waybill_no
        ,apply_type_code
        ,apply_time
from
    (
    select
    waybill_no
        ,apply_type_code
        ,apply_time
        ,row_number() over(partition by waybill_no order by apply_time desc,apply_type_code ) as rk  --优先退回件
    from jms_dwd.dwd_tab_reback_transfer_express_base
    where date(apply_time) between date_add('{{ execution_date | cst_ds }}',-60) and  '{{ execution_date | cst_ds }}'
    )t where rk=1
    )

insert overwrite table jms_dm.dm_download_sign_abnomal_detail_dt partition(dt)
select
    tt.billcode       -- 单号
     ,net.agent_code    --代理区code
     ,net.agent_name    --代理区name
     ,tt.store_name     --驿站nmae
     ,tt.store_code     --驿站code
     ,tt.site_code    --网点code
     ,net.name        as site_name    --网点name
     ,tt.scan_time_in --入库扫描时间
     ,tt.scan_time_out  --出库扫描时间
     ,tt.scan_time_take --快件取出扫描时间
     ,tt.signed_time    --签收扫描时间
     ,if(work_order.waybill_no  is not null,1,0) as is_work_order --是否客诉
     ,if(work_order.waybill_no  is not null and is_flas_sign=1,1,0) as is_flas_sign --是否虚假签收
     ,reback.apply_type_code --退转类型4退件5转寄
     ,tt.detail_type --导数类型
     ,case when tt.detail_type =4  then  1
           when tt.detail_type in(1,2,3) and sign_del.billcode is not null then  1
           else 0
    end as is_sign_del --是否签收删除
     ,delete_time
     ,date_time as date_time
     ,date_time as dt
from
    (   --签收异常
        select
            billcode,store_name,store_code ,scansitecode as site_code,scantime as scan_time_in,scan_time_out,scan_time_take,signed_time,signed_date as date_time,
            type1,type2,type3,p1.type_n as detail_type,null as delete_time
        from
            (
                select
                    b.billcode,b.store_name,b.store_code ,b.scansitecode ,b.scantime,b.scan_time_out,b.scan_time_take,a.signed_time,a.signed_date
                     ,case when a.signed_time>b.scantime       and b.scan_time_take is null       then 1  end as type1
                     ,case when a.signed_time>b.scan_time_out  and b.scan_time_out is not null    then 2  end as type2
                     ,case when a.signed_time<b.scantime       and b.scan_time_out is not null    then 3  end as type3
                from
                    (
                        select waybill_no ,max(signed_time) as signed_time --最近签收时间
                             ,date(max(signed_time)) as signed_date
                    from jms_dwd.dwd_tab_barscan_sign_base_dt
                where dt BETWEEN date_add('{{ execution_date | cst_ds }}',-6) and '{{ execution_date | cst_ds }}'
                  and date(signed_time) BETWEEN date_add('{{ execution_date | cst_ds }}',-6) and '{{ execution_date | cst_ds }}'
                group by waybill_no
            )a left join
            (
                select  billcode, scansitecode ,scantime,scan_time_out,scan_time_take,store_name,store_code,dt
                from jms_tmp.tmp_end_network_dispatch_in_detail_dt
                where dt between date_add('{{ execution_date | cst_ds }}',-8) and '{{ execution_date | cst_ds }}'
            )b on b.billcode=a.waybill_no  and b.dt between date_add(a.signed_date,-2) and a.signed_date --签收前3天
        where
            (a.signed_time>b.scantime      and b.scan_time_take is null)      ---有入库,但是没有取出操作 ,直接签收
           or (a.signed_time>b.scan_time_out and b.scan_time_out is not null) ---有出库,也有签收扫描，签收前有出库取3天
           or (a.signed_time<b.scantime      and b.scan_time_out is not null) ---签收扫描之后有入库扫描或者出库扫描
    )t lateral view explode (array(type1,type2,type3)) p1 as type_n
where p1.type_n is not null

union all
--签收删除
select
    a.billcode ,b.store_name,b.store_code,a.scan_site_code as site_code,b.scantime  as scan_time_in ,b.scan_time_out ,b.scan_time_take
     ,a.scan_time  as  signed_time --签收扫描时间
     ,a.date_time as date_time  --日期
     ,null as type1,null as type2,null as type3
     ,4 as detail_type
     ,a.delete_time    --签收删除时间
from
    (
    select billcode,scan_type,scan_time,scan_site_code,delete_time,date(delete_time) as  date_time
    from  jms_dwd.dwd_tab_barscan_sign_delete_dt
    where dt between date_add('{{ execution_date | cst_ds }}',-6) and '{{ execution_date | cst_ds }}'
    and date(delete_time) between date_add('{{ execution_date | cst_ds }}',-6) and '{{ execution_date | cst_ds }}'
    and scan_type='快件签收'
    group by billcode,scan_type,scan_time,scan_site_code,delete_time,date(delete_time)
    )a left join
    (
    select  billcode, scansitecode ,scantime,scan_time_out,scan_time_take,store_name,store_code
    from jms_tmp.tmp_end_network_dispatch_in_detail_dt
    where dt between date_add('{{ execution_date | cst_ds }}',-6) and '{{ execution_date | cst_ds }}'
    and scan_time_out is not null
    )b on b.billcode=a.billcode
    )tt left join jms_dim.dim_network_whole_massage net on net.code=tt.site_code
    left join work_order on work_order.waybill_no =tt.billcode and work_order.work_create_time>tt.signed_time
    left join reback on reback.waybill_no =tt.billcode and reback.apply_time>tt.signed_time
    left join
    (
select billcode,scan_type,scan_time,scan_site_code
from  jms_dwd.dwd_tab_barscan_sign_delete_dt
where dt between date_add('{{ execution_date | cst_ds }}',-6) and '{{ execution_date | cst_ds }}'
  and date(scan_time) between date_add('{{ execution_date | cst_ds }}',-6) and '{{ execution_date | cst_ds }}'
  and scan_type='快件签收'
    ) sign_del  on sign_del.billcode =tt.billcode and sign_del.scan_time=tt.signed_time and tt.detail_type in(1,2,3)
    distribute by pmod(hash(rand()), 3);

-----------------------------------------------------------------------------------------------------------
----汇总
insert overwrite table jms_dm.dm_download_sign_abnomal_cnt_dt partition(dt)
select
    agent_code                         --代理区code
     ,agent_name                         --代理区name
     ,sum(waybill_cnt1) as  waybill_cnt1 --导数类型1单量
     ,sum(waybill_cnt2) as  waybill_cnt2 --导数类型2单量
     ,sum(waybill_cnt3) as  waybill_cnt3 --导数类型3单量
     ,sum(waybill_cnt4) as  waybill_cnt4 --导数类型4单量
     ,sum(waybill_cnt1)+sum(waybill_cnt2)+sum(waybill_cnt3)+sum(waybill_cnt4)-max(distinct_cnt) as repeat_cnt  --重复单量
     ,max(is_work_order) as is_work_order --客诉量
     ,max(is_flas_sign)  as is_flas_sign  --客诉_虚假签收量
     ,max(sign_cnt) as  sign_cnt          --签收量
     ,dt as date_time                     --查询日期
     ,dt as dt                            --时间分区
from
    (
        select
            a.agent_code
             ,agent_name
             ,case when detail_type=1 then waybill_cnt else 0 end as waybill_cnt1
             ,case when detail_type=2 then waybill_cnt else 0 end as waybill_cnt2
             ,case when detail_type=3 then waybill_cnt else 0 end as waybill_cnt3
             ,case when detail_type=4 then waybill_cnt else 0 end as waybill_cnt4
             ,distinct_cnt
             ,is_work_order
             ,is_flas_sign
             ,s.sign_cnt
             ,a.dt
        from
            (
                SELECT
                    agent_code                --代理区编码
                     ,agent_name                --代理区名称
                     ,detail_type               --虚假签收类型明细
                     ,count(1) as  waybill_cnt  --各种类型单量
                     ,max(size(distinct_cnt))   as distinct_cnt   --异常签收单号去重量
                     ,max(size(is_work_order2)) as is_work_order  --客诉量
                     ,max(size(is_flas_sign2))  as is_flas_sign   --客诉_虚假签收量
                     ,dt
                FROM
                    (
                        select billcode, agent_code, agent_name, store_name, store_code, site_code, site_name, scan_time_in, scan_time_out, scan_time_take, signed_time, is_work_order, is_flas_sign, apply_type_code, detail_type, is_sign_del, dt,
                               collect_set( billcode) over(partition by agent_code,dt) as distinct_cnt   --所有类型单号去重单量
        ,collect_set( if(is_work_order=1,billcode,null)) over(partition by agent_code,dt) as is_work_order2 --单号去重_客诉量
        ,collect_set( if(is_flas_sign=1,billcode,null)) over(partition by agent_code,dt) as is_flas_sign2   --单号去重_虚假签收量
                        from jms_dm.dm_download_sign_abnomal_detail_dt
                        where dt between date_add('{{ execution_date | cst_ds }}',-6) and  '{{ execution_date | cst_ds }}'
                    )t1 group by agent_code,agent_name,detail_type,dt
            )a
                left join
            (   --每天签收量
                select
                    count(1) as sign_cnt --签收量
                     ,signed_time --签收时间
                     ,agent_code  --代理区编码
                from
                    (
                        select aa.waybill_no ,date(aa.signed_time) as signed_time,nn.agent_code
                    from jms_dwd.dwd_tab_barscan_sign_base_dt aa
            left join jms_dim.dim_network_whole_massage nn on nn.code=aa.signed_site_code
                where dt between date_add('{{ execution_date | cst_ds }}',-6) and '{{ execution_date | cst_ds }}'
                  and date(signed_time) between date_add('{{ execution_date | cst_ds }}',-6) and '{{ execution_date | cst_ds }}'
                group by aa.waybill_no ,date(aa.signed_time),nn.agent_code
            )a group by signed_time,agent_code
    )s on a.dt=s.signed_time and  a.agent_code=s.agent_code
)b group by agent_code,agent_name,dt
    distribute by 1 ;